package cn.xuexiyuan.flinkstudy.hello;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.Arrays;


/**
 * @Description: 使用 DataStream Lambda实现
 * @Author 左龙龙
 * @Date 21-3-18
 * @Version 1.0
 **/
public class WordCount5 {

    public static void main(String[] args) throws Exception {
        ParameterTool parameterTool = ParameterTool.fromArgs(args);

        // 0. env 创建 Flink 执行环境的上下文
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 使用 DataStream 根据数据源自动选择流还是批
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);

        // 1. source
        DataStream<String> lines = env.socketTextStream("localhost", 9999);

        // 2.transformation
        SingleOutputStreamOperator<String> words = lines.flatMap(
                (String line, Collector<String> out) -> Arrays.stream(line.toLowerCase().split("\\W+")).forEach(out::collect)
        ).returns(Types.STRING);

        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOne = words.map(
                (String word) -> Tuple2.of(word, 1)
        ).returns(Types.TUPLE(Types.STRING, Types.INT));

        // 分组
        KeyedStream<Tuple2<String, Integer>, String> grouped = wordAndOne.keyBy(t -> t.f0);
        // 聚合
        SingleOutputStreamOperator<Tuple2<String, Integer>> result = grouped.sum(1);

        // 3. sink
        // 如果执行 hdfs 权限相关错误，可以执行 hadoop fs -chmod -R 777 /
        // System.setProperty("HADOOP_USER_NAME", "ro0t");
        // 指定自定义参数实例：   -output ./output/
        if(parameterTool.has("output")){
          String output = parameterTool.get("output");
          result.writeAsText(output + System.currentTimeMillis()).setParallelism(1);
        }else{
            result.print();
        }

        //  4.excute 启动并等待程序结束
        env.execute("WordCount5");
    }

}
